package event;

import java.io.IOException;
import java.util.List;
import java.util.Random;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.util.Bytes;

public class DBAttrGenerator extends AttrGenerator{
	protected HTable table = null; 
	@Override
	public String generate(List<String> params) {
		initTables(params.get(0));
		int start, end;
		int rowNum = getRowNumber();
		if(params.size() > 1)
		{
			long curr = Long.parseLong(params.get(1));
			int split = Integer.parseInt(params.get(2));
			long total = Long.parseLong(params.get(3));
			long splitSize = total/split;
			int splitId = (int)(curr/splitSize);
			int rowPerSplit = rowNum/split;
			if (splitId < split - 1)
			{
				start = rowPerSplit * splitId;
				end = start + rowPerSplit;
			}
			else
			{
				start = rowPerSplit * (split - 1);
				end = rowNum;
			}
			System.out.println("table: "+ params.get(0) + " curr: " + curr + "  split: " + split + " total: " + total);
			System.out.println("start: "+ start + " end: " + end);
		}
		else
		{
			start = 0;
			end = rowNum;
		}
		Random random = new Random();
		int id  = start + random.nextInt(end - start);
		String name = getName(id);
		return name;
	}
	// get value from database with a rowkey and a column name
	private byte [] getValue(byte [] row, byte [] qualifier)
	{
		Get g = new Get(row);
		Result result;
		byte [] family = Bytes.toBytes(BasicInit.INFO);
		byte [] value = null;
		try {
			result = table.get(g);
			value = result.getValue(family, qualifier);
			return value;
		} catch (IOException e) {
			e.printStackTrace();
		}
		return value;
	}
	// Get the total number of rows in one database table
	private int getRowNumber()
	{
		byte [] row = Bytes.toBytes(BasicInit.TOTAL);
		byte [] qualifier = Bytes.toBytes(BasicInit.NUMBER);
		byte [] count = getValue(row, qualifier);
		return Bytes.toInt(count);
	}
	//get the value of the cell specified by rowkey and "Name" column 
	private String getName(int rowkey)
	{
		byte [] row = Bytes.toBytes(rowkey);
		byte [] qualifier = Bytes.toBytes(BasicInit.NAME);
		byte [] count = getValue(row, qualifier);
		return Bytes.toString(count);
	}
	private void initTables(String name)
	{
		Configuration config = HBaseConfiguration.create();
		try {
			byte []  tableName = Bytes.toBytes(name);
		    table = new HTable(config, tableName);
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}

}
